﻿using System;
using System.Collections;
using System.Collections.Generic;
using Me.Prettyprint.Cassandra.Connection;
using Me.Prettyprint.Cassandra.Serializers;
using Me.Prettyprint.Cassandra.Service;
using Me.Prettyprint.Cassandra.Utils;
using Me.Prettyprint.Hector.Api;
using Me.Prettyprint.Hector.Api.Beans;
using Me.Prettyprint.Hector.Api.Query;
using Me.Prettyprint.Hector.Api.Exceptions;
using Org.Apache.Cassandra.Thrift;

namespace Me.Prettyprint.Cassandra.Model
{
    /// <summary>
    /// HSlicePredicate is a slice of columns. There's a start and end for
    /// the slice.
    /// </summary>
    public enum PredicateType {Unknown, ColumnNames, Range};
    public class HSlicePredicate<N>
    {
        private static ByteBuffer EMPTY_BYTE_BUFFER = new ByteBuffer(new byte[0]);
        public HSlicePredicate(Serializer<N> columnNameSerializer)
        {
            if (columnNameSerializer == null)
            {
                throw new NullReferenceException("Column name serializer cannot be null");
            }
            ColumnNameSerializer = columnNameSerializer;
        }
        public List<N> ColumnNames
        {
            get;
            protected set;
        }
        protected N Start
        {
            get;set;
        }
        protected N Finish
        {
            get;set;
        }
        protected bool Reversed
        {
            get;set;
        }
        protected int Count
        {
            get;set;
        }
        protected bool CountSet
        {
            get;
            set;
        }
        public Serializer<N> ColumnNameSerializer
        {
            get;
            set;
        }
        protected PredicateType PredicateType
        {
            get;set;
        }
        public HSlicePredicate<N> SetColumnNames(List<N> columnNames)
        {
            ColumnNames = columnNames;
            PredicateType = PredicateType.ColumnNames;
            return this;
        }
        public HSlicePredicate<N> AddColumnName(N columnName) 
        {
            if (ColumnNames == null)
            {
                ColumnNames = new List<N>();
            }
            ColumnNames.Add(columnName);
            PredicateType = PredicateType.ColumnNames;
            return this;
        }
        public HSlicePredicate<N> SetKeysOnlyPredicate()
        {
            ColumnNames = new List<N>();
            PredicateType = PredicateType.ColumnNames;
            return this;
        }
        public HSlicePredicate<N> SetStartOn(N start)
        {
            Start = start;
            PredicateType = PredicateType.Range;
            return this;
        }
        public HSlicePredicate<N> SetEndOn(N finish)
        {
            Finish = finish;
            PredicateType = PredicateType.Range;
            return this;
        }
        public HSlicePredicate<N> SetCount(int count)
        {
            Count = count;
            CountSet = true;
            PredicateType = PredicateType.Range;
            return this;
        }
        public HSlicePredicate<N> SetReversed(bool reversed)
        {
            Reversed = reversed;
            PredicateType = PredicateType.Range;
            return this;
        }
        public HSlicePredicate<N> SetRange(N start, N finish, bool reversed, int count)
        {
            Start = start;
            Finish = finish;
            Reversed = reversed;
            Count = count;
            CountSet = true;
            PredicateType = PredicateType.Range;
            return this;
        }
        public List<N> GetColumnNames()
        {
            return ColumnNames;
        }
        public SlicePredicate ToThrift()
        {
            SlicePredicate pred = new SlicePredicate();
            switch (PredicateType)
            {
                case PredicateType.ColumnNames:
                    if (ColumnNames == null)
                    {
                        return null;
                    }
                    pred.Column_names = ToThriftColumnNames(ColumnNames);
                    pred.__isset.column_names = true;
                    break;
                case PredicateType.Range:
                    if (CountSet == false)
                    {
                        throw new NullReferenceException("Count was not set, neither were column-names set, can't execute");
                    }
                    SliceRange range = new SliceRange();
                    range.Start = FindBytes(Start);
                    range.Finish = FindBytes(Finish);
                    range.Reversed = Reversed;
                    range.Count = Count;
                    pred.Slice_range = range;
                    pred.__isset.slice_range = true;
                    break;
                case PredicateType.Unknown:
                default:
                    throw new HectorException(
                        "Neither column names nor range were set, this is an invalid slice predicate");
            }
            return pred;
        }
        private byte[] FindBytes(N val)
        {
            byte[] bytes = new byte[0];
            if (val != null)
            {
                bytes = ColumnNameSerializer.ToBytes(val);
            }
            return bytes;
        }
        private List<byte[]> ToThriftColumnNames(List<N> clms) 
        {
            List<byte[]> ret = new List<byte[]>(clms.Count);
            foreach (N name in clms)
            {
                ret.Add(ColumnNameSerializer.ToBytes(name));
            }
            return ret;
        }
        public override string ToString()
        {
            return string.Format("HSlicePredicate({0})", PredicateType == PredicateType.ColumnNames ? ColumnNames.ToString() : FormatPredicate());
        }

        private string FormatPredicate()
        {
            return string.Format("start:[{0}],end:[{1}],count:{2},reversed:{3}",
                Start != null ? Start.ToString() : "''",
                    Finish != null ? Finish.ToString() : "''",
                        Count, Reversed);
        }

    }
}
